package my.thesis.analytics;

import my.thesis.bolts.aggregators.HashtagCounter;
import my.thesis.bolts.aggregators.MentionsCounter;
import my.thesis.bolts.aggregators.SameLangHash;
import my.thesis.bolts.aggregators.SentimentCounter;
import my.thesis.bolts.aggregatorsReliable.HashtagCounterReliable;
import my.thesis.bolts.aggregatorsReliable.MentionsCounterReliable;
import my.thesis.bolts.aggregatorsReliable.SameLangHashReliable;
import my.thesis.bolts.aggregatorsReliable.SentimentCounterReliable;
import my.thesis.bolts.apicalls.CountryDetector;
import my.thesis.bolts.apicalls.CountryDetectorAsync;
import my.thesis.bolts.apicallsReliable.CountryDetectorReliable;
import my.thesis.bolts.computations.SentimentDetector;
import my.thesis.bolts.computations.TimeIntervalExtractor;
import my.thesis.bolts.computations.WordExtractor;
import my.thesis.bolts.computationsReliable.SentimentDetectorReliable;
import my.thesis.bolts.computationsReliable.TimeIntervalExtractorReliable;
import my.thesis.bolts.computationsReliable.WordExtractorReliable;
import my.thesis.bolts.fieldExtractors.TweetFieldExtractorJSON;
import my.thesis.bolts.fieldExtractors.TweetFieldExtractorTopo1;
import my.thesis.bolts.fieldExtractors.TweetFieldExtractorTopo2;
import my.thesis.bolts.fieldExtractorsReliable.TweetFieldExtractorJSONReliable;
import my.thesis.bolts.fieldExtractorsReliable.TweetFieldExtractorTopo1Reliable;
import my.thesis.bolts.fieldExtractorsReliable.TweetFieldExtractorTopo2Reliable;
import my.thesis.bolts.filters.FilterKeyword;
import my.thesis.bolts.filters.FilterLanguage;
import my.thesis.bolts.filtersReliable.FilterKeywordReliable;
import my.thesis.bolts.filtersReliable.FilterLanguageReliable;
import my.thesis.bolts.persistors.HashStreamPersistor;
import my.thesis.bolts.persistors.HashStreamPersistor2;
import my.thesis.bolts.persistors.InfluenceStreamPersistor;
import my.thesis.bolts.persistors.TextStreamPersistor;
import my.thesis.bolts.persistorsReliable.HashStreamPersistor2Reliable;
import my.thesis.bolts.persistorsReliable.HashStreamPersistorReliable;
import my.thesis.bolts.persistorsReliable.InfluenceStreamPersistorReliable;
import my.thesis.bolts.persistorsReliable.TextStreamPersistorReliable;
import my.thesis.spouts.HttpStream;
import my.thesis.spoutsReliable.HttpStreamReliable;
import my.thesis.test.TestBolt;

/*import storm.kafka.BrokerHosts;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import backtype.storm.spout.SchemeAsMultiScheme;*/

import backtype.storm.generated.StormTopology;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

public class TopologyBuilderUtil 
{
	 public static StormTopology buildTopo1(int timeIntervalSec,int timeToStoreHashtags,int timeToStoreCoOccurences
			 ,int fieldExtractorBoltParallelism,int fieldExtractorBoltNumTasks,int timeIntervalBoltParallelism,int timeIntervalBoltNumTasks,
			 int counterParallelism, int counterNumTasks,int counterParallelism2, int counterNumTasks2, String webserver,String database)
	 {
	  		TopologyBuilder builder = new TopologyBuilder();
	      	   
	  	     builder.setSpout("HttpStreamSpout", new HttpStream(webserver),1);
	    	
	    	 builder
	   	   	.setBolt("TweetFieldExtractorTopo1Bolt", new TweetFieldExtractorTopo1(),fieldExtractorBoltParallelism)
	   	   	.setNumTasks(fieldExtractorBoltNumTasks)
	   	   	.shuffleGrouping("HttpStreamSpout");

	    	 builder
	     	.setBolt("TimeIntervalExtractorBolt", new TimeIntervalExtractor(timeIntervalSec),timeIntervalBoltParallelism)
	     	.setNumTasks(timeIntervalBoltNumTasks)
	     	.shuffleGrouping("TweetFieldExtractorTopo1Bolt");
	    	 
			 builder
			.setBolt("HashtagCounterBolt", new HashtagCounter(timeToStoreHashtags),counterParallelism)
			.setNumTasks(counterNumTasks)
		    .fieldsGrouping("TimeIntervalExtractorBolt",new Fields("timeInterval","lang"));
								    	    	 
			builder
			.setBolt("HashStreamPersistorBolt", new HashStreamPersistor(timeIntervalSec,database),1)
			.setNumTasks(1)
			.globalGrouping("HashtagCounterBolt");
								    		  	     	    	 
			builder
			.setBolt("SameLangHashBolt", new SameLangHash(timeToStoreCoOccurences),counterParallelism)
			.setNumTasks(counterNumTasks)
			.fieldsGrouping("TimeIntervalExtractorBolt",new Fields("timeInterval"));
								    	       	 
			builder
			.setBolt("HashStreamPersistor2Bolt", new HashStreamPersistor2(timeIntervalSec,database),1)
			.setNumTasks(1)
			.globalGrouping("SameLangHashBolt");
								    	       	 
			return builder.createTopology();
	    	       	 
	}
	 
	public static StormTopology buildTopo2(int timeToStoreSentiment,int timeToStoreMention
			 ,int fieldExtractorBoltParallelism,int fieldExtractorBoltNumTasks, int filterParallelism, int filterNumTasks,
			 int countryDetectorBoltParallelism,int countryDetectorBoltNumTasks, int counterParallelism, int counterNumTasks,
			 int counterParallelism2, int counterNumTasks2,int wordExtractParallelism, int wordExtractNumTasks,int sentimentParallelism,int sentimentNumTasks,
			 String webserver,String database)
	 {
	  		TopologyBuilder builder = new TopologyBuilder();
	      	   
	  	     builder.setSpout("HttpStreamSpout", new HttpStream(webserver),1);
	    	
	    	 builder
	   	   	.setBolt("TweetFieldExtractorTopo2Bolt", new TweetFieldExtractorTopo2(),fieldExtractorBoltParallelism)
	   	   	.setNumTasks(fieldExtractorBoltNumTasks)
	   	   	.shuffleGrouping("HttpStreamSpout");
	    	 
	    	 builder
 	    	.setBolt("FilterLanguage", new FilterLanguage(),filterParallelism)
 	    	.setNumTasks(filterNumTasks)
 	    	.shuffleGrouping("TweetFieldExtractorTopo2Bolt","textStream");
 		  	
 		  	 builder
 	    	.setBolt("CountryDetectorBolt", new CountryDetector(),countryDetectorBoltParallelism)
 	    	.setNumTasks(countryDetectorBoltNumTasks)
 	    	.shuffleGrouping("FilterLanguage");
 	    	  	
 	      	 builder
 	    	.setBolt("WordExtractorBolt", new WordExtractor(),wordExtractParallelism)
 	    	.setNumTasks(wordExtractNumTasks)
 	    	.shuffleGrouping("CountryDetectorBolt");
 	      	 
 	      	 builder
 	    	.setBolt("SentimentDetectorBolt", new SentimentDetector(),sentimentParallelism)
 	    	.setNumTasks(sentimentNumTasks)
 	    	.shuffleGrouping("WordExtractorBolt");
 	    	
 	    	 builder
 	    	.setBolt("SentimentCounterBolt", new SentimentCounter(timeToStoreSentiment),counterParallelism)
 	    	.setNumTasks(counterNumTasks)
 	    	.fieldsGrouping("SentimentDetectorBolt",new Fields("country"));
 	     	
 	       	 builder
 	    	.setBolt("SentimentPersistorBolt", new TextStreamPersistor(database),1)
 	    	.setNumTasks(1)
 	    	.globalGrouping("SentimentCounterBolt");
 	       	 
 	       	 builder
	     	.setBolt("FilterKeywordBolt", new FilterKeyword(),filterParallelism)
	     	.setNumTasks(filterNumTasks)
	     	.shuffleGrouping("TweetFieldExtractorTopo2Bolt","influenceStream");

	    	 builder
	     	.setBolt("MentionCounterBolt", new MentionsCounter(timeToStoreMention),counterParallelism2)
	     	.setNumTasks(counterNumTasks2)
	     	.shuffleGrouping("FilterKeywordBolt");
	    	 
	    	 builder
	      	.setBolt("MentionsPersistorBolt", new InfluenceStreamPersistor(database),1)
	      	.setNumTasks(1)
	      	.globalGrouping("MentionCounterBolt");   	    	 

	    	 return builder.createTopology();
	    	 			    	       	 	       	 
	}
	
	public static StormTopology buildTopo1Reliable(int timeIntervalSec,int timeToStoreHashtags,int timeToStoreCoOccurences
			 ,int fieldExtractorBoltParallelism,int fieldExtractorBoltNumTasks,int timeIntervalBoltParallelism,int timeIntervalBoltNumTasks,
			 int counterParallelism, int counterNumTasks,int counterParallelism2, int counterNumTasks2,String webserver,String database)
	 {
	  		TopologyBuilder builder = new TopologyBuilder();
	      	   
	  	     builder.setSpout("HttpStreamSpout", new HttpStreamReliable(webserver),1);
	    	
	    	 builder
	   	   	.setBolt("TweetFieldExtractorTopo1Bolt", new TweetFieldExtractorTopo1Reliable(),fieldExtractorBoltParallelism)
	   	   	.setNumTasks(fieldExtractorBoltNumTasks)
	   	   	.shuffleGrouping("HttpStreamSpout");

	    	 builder
	     	.setBolt("TimeIntervalExtractorBolt", new TimeIntervalExtractorReliable(timeIntervalSec),timeIntervalBoltParallelism)
	     	.setNumTasks(timeIntervalBoltNumTasks)
	     	.shuffleGrouping("TweetFieldExtractorTopo1Bolt");
	    	 
			 builder
			.setBolt("HashtagCounterBolt", new HashtagCounterReliable(timeToStoreHashtags),counterParallelism)
			.setNumTasks(counterNumTasks)
		    .fieldsGrouping("TimeIntervalExtractorBolt",new Fields("timeInterval","lang"));
								    	    	 
			builder
			.setBolt("HashStreamPersistorBolt", new HashStreamPersistorReliable(timeIntervalSec,database),1)
			.setNumTasks(1)
			.globalGrouping("HashtagCounterBolt");
								    		  	 
						    	    	 
			builder
			.setBolt("SameLangHashBolt", new SameLangHashReliable(timeToStoreCoOccurences),counterParallelism2)
			.setNumTasks(counterNumTasks2)
			.fieldsGrouping("TimeIntervalExtractorBolt",new Fields("timeInterval"));
								    	       	 
			builder
			.setBolt("HashStreamPersistor2Bolt", new HashStreamPersistor2Reliable(timeIntervalSec,database),1)
			.setNumTasks(1)
			.globalGrouping("SameLangHashBolt");
								    	       	 
			return builder.createTopology();
	    	       	 
	}
	 
	public static StormTopology buildTopo2Reliable(int timeToStoreSentiment,int timeToStoreMention
			 ,int fieldExtractorBoltParallelism,int fieldExtractorBoltNumTasks, int filterParallelism, int filterNumTasks,
			 int countryDetectorBoltParallelism,int countryDetectorBoltNumTasks, int counterParallelism, int counterNumTasks,int counterParallelism2, int counterNumTasks2,
			 String webserver,String database)
	 {
	  		TopologyBuilder builder = new TopologyBuilder();
	      	   
	  	     builder.setSpout("HttpStreamSpout", new HttpStreamReliable(webserver),1);
	    	
	    	 builder
	   	   	.setBolt("TweetFieldExtractorTopo2Bolt", new TweetFieldExtractorTopo2Reliable(),fieldExtractorBoltParallelism)
	   	   	.setNumTasks(fieldExtractorBoltNumTasks)
	   	   	.shuffleGrouping("HttpStreamSpout");
	    	 
	    	 builder
	    	.setBolt("FilterLanguage", new FilterLanguageReliable(),filterParallelism)
	    	.setNumTasks(filterNumTasks)
	    	.shuffleGrouping("TweetFieldExtractorTopo2Bolt","textStream");
		  	
		  	 builder
	    	.setBolt("CountryDetectorBolt", new CountryDetectorReliable(),countryDetectorBoltParallelism)
	    	.setNumTasks(countryDetectorBoltNumTasks)
	    	.shuffleGrouping("FilterLanguage");
	    	  	
	      	 builder
	    	.setBolt("WordExtractorBolt", new WordExtractorReliable(),1)
	    	.setNumTasks(1)
	    	.shuffleGrouping("CountryDetectorBolt");
	      	 
	      	 builder
	    	.setBolt("SentimentDetectorBolt", new SentimentDetectorReliable(),1)
	    	.setNumTasks(1)
	    	.shuffleGrouping("WordExtractorBolt");
	    	
	    	 builder
	    	.setBolt("SentimentCounterBolt", new SentimentCounterReliable(timeToStoreSentiment),counterParallelism)
	    	.setNumTasks(counterNumTasks)
	    	.fieldsGrouping("SentimentDetectorBolt",new Fields("country"));
	     	
	       	 builder
	    	.setBolt("SentimentPersistorBolt", new TextStreamPersistorReliable(database),1)
	    	.setNumTasks(1)
	    	.globalGrouping("SentimentCounterBolt");
	       	 
	       	 builder
	     	.setBolt("FilterKeywordBolt", new FilterKeywordReliable(),1)
	     	.setNumTasks(1)
	     	.shuffleGrouping("TweetFieldExtractorTopo2Bolt","influenceStream");

	    	 builder
	     	.setBolt("MentionCounterBolt", new MentionsCounterReliable(timeToStoreMention),counterParallelism2)
	     	.setNumTasks(counterNumTasks2)
	     	.shuffleGrouping("FilterKeywordBolt");
	    	 
	    	 builder
	      	.setBolt("MentionsPersistorBolt", new InfluenceStreamPersistorReliable(database),1)
	      	.setNumTasks(1)
	      	.globalGrouping("MentionCounterBolt");   	    	 

	    	 return builder.createTopology();
	    	 			    	       	 	       	 
	} 

}
